package com.hps.shoping.order.component;

import com.alibaba.fastjson.JSON;
import com.hps.shoping.order.entitys.Order;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.nio.charset.Charset;
import java.util.concurrent.Executors;

/**
 * @author heps
 * @date 2019/9/1 10:00
 */
@Component
@Slf4j
public class MessageProducer {

    TransactionMQProducer producer;

    public MessageProducer() {
        producer = new TransactionMQProducer("order_group");
        producer.setNamesrvAddr("192.168.117.139:9876");
        producer.setExecutorService(Executors.newFixedThreadPool(10));
        producer.setTransactionListener(new TransactionListenerImpl());

        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    private static class TransactionListenerImpl implements TransactionListener {

        @Override
        public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            log.info("execute order trans {}", arg);
            return LocalTransactionState.UNKNOW;
        }

        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            String body = new String(msg.getBody(), Charset.defaultCharset());
            Order order = JSON.parseObject(body, Order.class);
            log.info("check order transaction {}", order);
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }

    public void send(String message) throws Exception {
        Message msg = new Message("orderTopic", message.getBytes(Charset.defaultCharset()));
        SendResult sendResult = producer.sendMessageInTransaction(msg, null);
        log.info("send order msg {}", sendResult);
    }
}
